package com.aoyama.kafka.base.demo03;

import com.aoyama.kafka.config.KafkaInitialConfiguration;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;

/**
 * 重点：消费者各个参数设置说明
 *
 * ① id：消费者ID；
 *
 * ② groupId：消费组ID；
 *
 * ③ topics：监听的topic，可监听多个；
 *
 * ④ topicPartitions：可配置更加详细的监听信息，可指定topic、parition、offset监听。
 *
 */

@Component
public class KafkaArgsConsumer01 {

    /**
     * 首先可根据{@link KafkaInitialConfiguration}生产topic3 和 4 （注意让分区数，分区过多，可能导致消费时消费不到数据）
     * 测试推送数据{@link KafkaArgsConsumerTest#kafkaArgsConsumer()}
     *
     * 指定topic、partition、offset消费
     * 同时监听topic1和topic2，监听topic1的0号分区、topic2的 "0号和1号" 分区，指向1号分区的offset初始值为8
     * @param record
     */
    @KafkaListener(id = "consumer1",topicPartitions = {
            @TopicPartition(topic = "topic3", partitions = { "0" }),
            @TopicPartition(topic = "topic4", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "3"))
    })
    public void onMessage(ConsumerRecord<?, ?> record) {
        System.out.println("-----88888----topic:"+record.topic()+"|partition:"+record.partition()+"|offset:"+record.offset()+"|value:"+record.value());
    }
}
